package com.ds.infra.mq.example;


import com.ds.infra.mq.core.api.MQConsumer;

import java.util.Scanner;

public class ConsumerExample {
    public static void main(String[] args) throws Exception {
        MQConsumer consumer = new MQConsumer("localhost", 9092, "consumer-" + System.currentTimeMillis());
        consumer.connect();

        String topic = Constant.TEST_TOPIC;
        Scanner scanner = new Scanner(System.in);
        int partition = Constant.PARTITION;
        System.out.println("Enter partition (default: 0):");
        try {
            String partitionStr = scanner.nextLine().trim();
            if (!partitionStr.isEmpty()) {
                partition = Integer.parseInt(partitionStr);
            }
        } catch (NumberFormatException e) {
            System.out.println("Invalid partition, using default 0");
        }

        System.out.println("Consumer started. Waiting for messages...");

        while (true) {
            try {
                byte[] messageBytes = consumer.poll(topic, partition);
                if (messageBytes != null) {
                    System.out.println(new String(messageBytes));
                } else {
                    // 空响应时适当等待
                    Thread.sleep(1000);
                }
            } catch (Exception e) {
                System.err.println("Consume error: " + e.getMessage());
                Thread.sleep(5000); // 错误后等待
            }
        }
    }
}
